package com.atguigu.udmp.stream.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SimplePropertyPreFilter;
import com.atguigu.udmp.stream.bean.DimInfo;
import com.atguigu.udmp.stream.bean.DimInfoDefine;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Date;
import java.util.Map;

public class DimInfoExtractProcessFunction extends BroadcastProcessFunction<JSONObject, JSONObject, DimInfo> {

    MapStateDescriptor<Long,DimInfoDefine> dimInfoDefineMapStateDescriptor;


    public DimInfoExtractProcessFunction(MapStateDescriptor<Long, DimInfoDefine> dimInfoDefineMapStateDescriptor){
        this.dimInfoDefineMapStateDescriptor=dimInfoDefineMapStateDescriptor;
    }

    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, JSONObject, DimInfo>.ReadOnlyContext readOnlyContext, Collector<DimInfo> collector) throws Exception {
        ReadOnlyBroadcastState<Long, DimInfoDefine> broadcastState = readOnlyContext.getBroadcastState(dimInfoDefineMapStateDescriptor);

        //遍历有的维度定义，找到对应匹配
        for (Map.Entry<Long, DimInfoDefine> dimInfoDefineEntry : broadcastState.immutableEntries()) {
            DimInfoDefine dimInfoDefine = dimInfoDefineEntry.getValue();
            //调用抽取方法
            DimInfo dimInfo = extract(jsonObject, dimInfoDefine);
            collector.collect(dimInfo);
        }
    }

    @Override
    public void processBroadcastElement(JSONObject dimDefineDataJsonObj, BroadcastProcessFunction<JSONObject, JSONObject, DimInfo>.Context context, Collector<DimInfo> collector) throws Exception {

        // 获得定义信息
        BroadcastState<Long, DimInfoDefine> broadcastState = context.getBroadcastState(dimInfoDefineMapStateDescriptor);

        // 获得操作类型
        String opType=dimDefineDataJsonObj.getString("op");
        JSONObject dataAfterJsonObj = dimDefineDataJsonObj.getJSONObject("after");
        DimInfoDefine dimInfoDefine = dataAfterJsonObj.toJavaObject(DimInfoDefine.class);

        //
        // 处理c 创建 , r 预加载, u 修改
        if(opType.equals("c")||opType.equals("r")||opType.equals("u")){
            // 从jsonObj中提取数据 加载到userEventDefine
            try{
                broadcastState.put(dimInfoDefine.getId(),dimInfoDefine);
            }catch (Exception e){
                throw  new RuntimeException("维度数据生成有误");
            }

        } else if (opType.equals("d")) {
            broadcastState.remove(dimInfoDefine.getId() );
        }
    }

    private DimInfo extract(JSONObject dataJsonObj, DimInfoDefine dimInfoDefine){
        DimInfo dimInfo = new DimInfo();

        JSONObject afterJsonObj = dataJsonObj.getJSONObject("after");
        //提取内容
        dimInfo.setDimId(afterJsonObj.getLong("id"));
        dimInfo.setDimDefineId(dimInfoDefine.getId());
        dimInfo.setDimDefineName(dimInfoDefine.getDimName());
        dimInfo.setUpsertTime(DateFormatUtils.format(new Date(),"yyyy-MM-dd HH:mm:ss"));

        //过滤json的内容
        String sinkColumns = dimInfoDefine.getSinkColumns();
        String[] columnArr = sinkColumns.split(",");
        SimplePropertyPreFilter simplePropertyPreFilter = new SimplePropertyPreFilter(columnArr);
        String dimInfoJson = JSON.toJSONString(afterJsonObj, simplePropertyPreFilter);
        dimInfo.setDimInfoJson(dimInfoJson);

        return  dimInfo;
    }
}
